package com.easepal.socketiolib.rxbus;

import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * 作者: bright.lin@medisana.net.cn
 * 时间: 2018/10/23
 * 标题: RxBus事件总线
 * 描述: 用于处理接收和发送各种事件（不支持背压）-基于RxJava框架
 * 修改: 由 bright.lin@medisana.net.cn 修改于 2018/10/23
 */
@SuppressWarnings("unused")
public class RxBus {

    // 事件总线核心（不支持背压）
    private final Subject<Object> mBus;

    /**
     * 创建mBus
     */
    private RxBus() {
        mBus = PublishSubject.create().toSerialized();
    }

    /**
     * 获取RxBus单例
     *
     * @return RxBus单例
     */
    public static RxBus get() {
        return Holder.BUS;
    }

    /**
     * 发送事件
     *
     * @param o Object对象
     */
    public void post(Object o) {
        mBus.onNext(o);
    }

    /**
     * 观察事件
     *
     * @param tClass 事件类
     * @param <T>    泛型
     * @return Observable泛型对象
     */
    public <T> Observable<T> toObservable(Class<T> tClass) {
        return mBus.ofType(tClass);
    }

    /**
     * 观察事件
     *
     * @return Observable对象
     */
    public Observable<Object> toObservable() {
        return mBus;
    }

    /**
     * 是否有观察者
     *
     * @return true/false
     */
    public boolean hasObservers() {
        return mBus.hasObservers();
    }

    /**
     * 注册事件监听
     *
     * @param tClass    事件类
     * @param scheduler 被观察线程
     * @param onNext    接收事件
     * @param <T>       泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Scheduler scheduler, Consumer<T> onNext) {
        return toObservable(tClass).observeOn(scheduler).subscribe(onNext);
    }

    /**
     * 注册事件监听
     *
     * @param tClass    事件类
     * @param scheduler 被观察线程
     * @param onNext    接收事件
     * @param onError   错误处理
     * @param <T>       泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Scheduler scheduler, Consumer<T> onNext, Consumer<Throwable> onError) {
        return toObservable(tClass).observeOn(scheduler).subscribe(onNext, onError);
    }

    /**
     * 注册事件监听
     *
     * @param tClass     事件类
     * @param scheduler  被观察线程
     * @param onNext     接收事件
     * @param onError    错误处理
     * @param onComplete 完成处理
     * @param <T>        泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Scheduler scheduler, Consumer<T> onNext, Consumer<Throwable> onError, Action onComplete) {
        return toObservable(tClass).observeOn(scheduler).subscribe(onNext, onError, onComplete);
    }

    /**
     * 注册事件监听
     *
     * @param tClass      事件类
     * @param scheduler   被观察线程
     * @param onNext      接收事件
     * @param onError     错误处理
     * @param onComplete  完成处理
     * @param onSubscribe 观察线程
     * @param <T>         泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Scheduler scheduler, Consumer<T> onNext, Consumer<Throwable> onError, Action onComplete, Consumer<Disposable> onSubscribe) {
        return toObservable(tClass).observeOn(scheduler).subscribe(onNext, onError, onComplete, onSubscribe);
    }

    /**
     * 注册事件监听(在主线程被观察）
     *
     * @param tClass 事件类
     * @param onNext 接收事件
     * @param <T>    泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Consumer<T> onNext) {
        return toObservable(tClass).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext);
    }

    /**
     * 注册事件监听(在主线程被观察）
     *
     * @param tClass  事件类
     * @param onNext  接收事件
     * @param onError 错误处理
     * @param <T>     泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Consumer<T> onNext, Consumer<Throwable> onError) {
        return toObservable(tClass).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext, onError);
    }

    /**
     * 注册事件监听(在主线程被观察）
     *
     * @param tClass     事件类
     * @param onNext     接收事件
     * @param onError    错误处理
     * @param onComplete 完成处理
     * @param <T>        泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Consumer<T> onNext, Consumer<Throwable> onError, Action onComplete) {
        return toObservable(tClass).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext, onError, onComplete);
    }

    /**
     * 注册事件监听(在主线程被观察）
     *
     * @param tClass      事件类
     * @param onNext      接收事件
     * @param onError     错误处理
     * @param onComplete  完成处理
     * @param onSubscribe 观察线程
     * @param <T>         泛型
     * @return Disposable泛型对象
     */
    public <T> Disposable register(Class<T> tClass, Consumer<T> onNext, Consumer<Throwable> onError, Action onComplete, Consumer<Disposable> onSubscribe) {
        return toObservable(tClass).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext, onError, onComplete, onSubscribe);
    }

    /**
     * 注销事件监听
     *
     * @param disposable Disposable对象
     */
    public void unregister(Disposable disposable) {
        if (null != disposable && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }

    /**
     * 单例Holder
     */
    private static class Holder {
        private static final RxBus BUS = new RxBus();
    }
}
